package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.DataFrame

/**
  * @Author songhaicheng
  * @Date 2018/10/23 19:22
  * @Description
  * @Reviewer
  */
case class VectorAssembleEncoderParam(
                                       override val input_pt: String,
                                       override val output_pt: String,
                                       override val hive_table: String,
                                       override val flow_time: String,
                                       override val inputCol: String,
                                       override val outputCol: String,
                                       override val preserveCols: String,
                                       override val modelPath: String
                                     ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null)

}


class VectorAssembleEncoder extends AbstractFeatureEncoder[VectorAssembleEncoderParam] {

  override def setUp(param: VectorAssembleEncoderParam, data: DataFrame): Array[PipelineStage] = {
    val vs = new VectorAssembler()
      .setInputCols(param.inputCol.split(","))
      .setOutputCol(param.outputCol)

    Array(vs)
  }

}

object VectorAssembleEncoder {

  def apply(paramStr: String): Unit = {
    new VectorAssembleEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    VectorAssembleEncoder(args(0))
  }

}
